package com.atguigu.gmall.realtime.app

import java.time.{LocalDate, Period}
import java.util

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.zxq.gmall.realtime.util.MyEsUtils
import com.zxq.mall.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.zxq.mall.realtime.util.{MyKafkaUtils, MyRedisUtils, OffsetManagerUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/
object DwdOrderApp {

  def main(args: Array[String]): Unit = {

    val sparconf: SparkConf = new SparkConf().setAppName("dwd_order_app").setMaster("local[4]")
    val ssc: StreamingContext = new StreamingContext(sparconf , Seconds(5))


    val orderInfoTopicName : String = "DWD_ORDER_INFO_I"
    val orderInfoGroup : String = "DWD_ORDER_INFO:GROUP"
    val orderInfoOffsets: Map[TopicPartition, Long] =
      OffsetManagerUtil.getOffset(orderInfoTopicName , orderInfoGroup)

    val orderDetailTopicName : String = "DWD_ORDER_DETAIL_I"
    val orderDetailGroup : String = "DWD_ORDER_DETAIL_GROUP"
    val orderDetailOffsets: Map[TopicPartition, Long] =
      OffsetManagerUtil.getOffset(orderDetailTopicName ,orderDetailGroup)


    var orderInfoKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(orderInfoOffsets != null && orderInfoOffsets.nonEmpty){
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(orderInfoTopicName,ssc, orderInfoOffsets,orderInfoGroup)
    }else{
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(ssc,orderInfoTopicName, orderInfoGroup)
    }


    var orderDetailKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(orderDetailOffsets!= null && orderDetailOffsets.nonEmpty){
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(orderDetailTopicName,ssc,orderDetailOffsets,orderDetailGroup)
    }else{
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(ssc,orderDetailTopicName,orderDetailGroup)
    }


    var orderInfoOffsetRanges: Array[OffsetRange] = null
    val orderInfoOffsetDStream: DStream[ConsumerRecord[String, String]] = orderInfoKafkaDStream.transform(
      rdd => {
        orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )


    var  orderDetailOffsetRanges: Array[OffsetRange] = null
    val orderDetailOffsetDStream: DStream[ConsumerRecord[String, String]] = orderDetailKafkaDStream.transform(
      rdd => {
        orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )


    val orderInfoDStream: DStream[OrderInfo] = orderInfoOffsetDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val orderInfo: OrderInfo = JSON.parseObject(value, classOf[OrderInfo])
        orderInfo
      }
    )


    val orderDetailDStream: DStream[OrderDetail] = orderDetailOffsetDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val orderDetail: OrderDetail = JSON.parseObject(value, classOf[OrderDetail])
        orderDetail
      }
    )

    val orderInfoDimDStream: DStream[OrderInfo] = orderInfoDStream.mapPartitions(
      orderInfoIter => {

        val orderInfos: List[OrderInfo] = orderInfoIter.toList
        val jedis: Jedis = MyRedisUtils.getJedisClient
        for (orderInfo <- orderInfos) {

          val uid: Long = orderInfo.user_id
          val redisUserKey: String = s"DIM:USER_INFO:$uid"
          val userInfoJson: String = jedis.get(redisUserKey)
          val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)

          val gender: String = userInfoJsonObj.getString("gender")

          val birthday: String = userInfoJsonObj.getString("birthday")

          val birthdayLd: LocalDate = LocalDate.parse(birthday)
          val nowLd: LocalDate = LocalDate.now()
          val period: Period = Period.between(birthdayLd, nowLd)
          val age: Int = period.getYears


          orderInfo.user_gender = gender
          orderInfo.user_age = age



          val provinceID: Long = orderInfo.province_id
          val redisProvinceKey: String = s"DIM:BASE_PROVINCE:$provinceID"
          val provinceJson: String = jedis.get(redisProvinceKey)
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)

          val provinceName: String = provinceJsonObj.getString("name")
          val provinceAreaCode: String = provinceJsonObj.getString("area_code")
          val province3166: String = provinceJsonObj.getString("iso_3166_2")
          val provinceIsoCode: String = provinceJsonObj.getString("iso_code")


          orderInfo.province_name = provinceName
          orderInfo.province_area_code = provinceAreaCode
          orderInfo.province_3166_2_code = province3166
          orderInfo.province_iso_code = provinceIsoCode


          val createTime: String = orderInfo.create_time
          val createDtHr: Array[String] = createTime.split(" ")
          val createDate: String = createDtHr(0)
          val createHr: String = createDtHr(1).split(":")(0)

          orderInfo.create_date = createDate
          orderInfo.create_hour = createHr


        }
        jedis.close()
        orderInfos.iterator
      }
    )

    val orderInfoKVDStream: DStream[(Long, OrderInfo)] =
    orderInfoDimDStream.map( orderInfo => (orderInfo.id , orderInfo))

    val orderDetailKVDStream: DStream[(Long, OrderDetail)] =
      orderDetailDStream.map(orderDetail => (orderDetail.order_id , orderDetail))


    val orderJoinDStream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] =
    orderInfoKVDStream.fullOuterJoin(orderDetailKVDStream)

    val orderWideDStream: DStream[OrderWide] = orderJoinDStream.mapPartitions(
      orderJoinIter => {
        val jedis: Jedis = MyRedisUtils.getJedisClient
        val orderWides: ListBuffer[OrderWide] = ListBuffer[OrderWide]()
        for ((key, (orderInfoOp, orderDetailOp)) <- orderJoinIter) {

          if (orderInfoOp.isDefined) {

            val orderInfo: OrderInfo = orderInfoOp.get
            if (orderDetailOp.isDefined) {

              val orderDetail: OrderDetail = orderDetailOp.get

              val orderWide: OrderWide = new OrderWide(orderInfo, orderDetail)

              orderWides.append(orderWide)
            }

            val redisOrderInfoKey: String = s"ORDERJOIN:ORDER_INFO:${orderInfo.id}"

            jedis.setex(redisOrderInfoKey, 24 * 3600, JSON.toJSONString(orderInfo, new SerializeConfig(true)))


            val redisOrderDetailKey: String = s"ORDERJOIN:ORDER_DETAIL:${orderInfo.id}"
            val orderDetails: util.Set[String] = jedis.smembers(redisOrderDetailKey)
            if (orderDetails != null && orderDetails.size() > 0) {
              import scala.collection.JavaConverters._
              for (orderDetailJson <- orderDetails.asScala) {
                val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])

                val orderWide: OrderWide = new OrderWide(orderInfo, orderDetail)

                orderWides.append(orderWide)
              }
            }

          } else {

            val orderDetail: OrderDetail = orderDetailOp.get

            val redisOrderInfoKey: String = s"ORDERJOIN:ORDER_INFO:${orderDetail.order_id}"
            val orderInfoJson: String = jedis.get(redisOrderInfoKey)
            if (orderInfoJson != null && orderInfoJson.nonEmpty) {
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])

              val orderWide: OrderWide = new OrderWide(orderInfo, orderDetail)

              orderWides.append(orderWide)
            } else {

              val redisOrderDetailKey: String = s"ORDERJOIN:ORDER_DETAIL:${orderDetail.order_id}"
              jedis.sadd(redisOrderDetailKey, JSON.toJSONString(orderDetail, new SerializeConfig(true)))
              jedis.expire(redisOrderDetailKey, 24 * 3600)
            }
          }
        }
        jedis.close()
        orderWides.iterator
      }
    )

    orderWideDStream.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          orderWideIter => {
            val orderWides: List[(String, OrderWide)] =
              orderWideIter.map( orderWide => (orderWide.detail_id.toString , orderWide)).toList
            if(orderWides.size > 0 ){
              val head: (String, OrderWide) = orderWides.head
              val date: String = head._2.create_date

              val indexName : String = s"gmall_order_wide_$date"

              MyEsUtils.bulkSave(orderWides ,indexName )
            }
          }
        )

        OffsetManagerUtil.saveOffset(orderInfoTopicName , orderInfoGroup , orderInfoOffsetRanges)
        OffsetManagerUtil.saveOffset(orderDetailTopicName , orderDetailGroup ,orderDetailOffsetRanges)
      }
    )


    ssc.start()
    ssc.awaitTermination()
  }
}

